package org.apache.ocean.main;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.ocean.main.util.Timeout;

public class TransactionMonitor<K,P> {
	public static enum State {SUBMITTED, PREPARED, COMMITTED, FAILED};
  private Map<K,Transaction<P>> map;
  
  public TransactionMonitor() {
  	map = new HashMap<K,Transaction<P>>();
  }
  
  public void add(K key, Set<P> participants) throws TransactionException {
  	Transaction<P> transaction = new Transaction<P>(participants);
  	if (map.containsKey(key)) {
  		throw new TransactionException("contains key: "+key);
  	}
  	map.put(key, transaction);
  }
  
  public static class TransactionException extends Exception {
  	public TransactionException(String message) {
  		super(message);
  	}
  }
  
  public void waitPrepare(K key, P participant, Timeout timeout) throws TransactionException, TimeoutException, InterruptedException {
  	Transaction<P> transaction = map.get(key);
  	if (transaction == null) {
  		throw new TransactionException("unknown transaction key: "+key);
  	}
  	transaction.waitForPrepared(participant, timeout);
  }
  
  public void waitCommit(K key, P participant, Timeout timeout) throws TransactionException, TimeoutException, InterruptedException {
  	Transaction<P> transaction = map.get(key);
  	if (transaction == null) {
  		throw new TransactionException("unknown transaction key: "+key);
  	}
  	transaction.waitForCommit(participant, timeout);
  }
  
  public static class Transaction<P> {
  	private CountDownLatch latch;
  	private Set<P> participants;
  	private ReentrantLock lock = new ReentrantLock();
  	private Condition commitCondition = lock.newCondition();
  	private AtomicReference<State> stateRef = new AtomicReference<State>();
  	private List errors = new ArrayList();
  	
  	public Transaction(Set<P> participants) {
  		latch = new CountDownLatch(participants.size());
  	}
  	
  	public void addError(Object error) {
  		errors.add(error);
  	}
  	
  	private void throwTransactionException(String message) throws TransactionException {
  		// TODO: include transaction toString
  		throw new TransactionException(message);
  	}
  	
  	private void checkErrors() throws TransactionException {
  		if (errors.size() > 0) {
  			throwTransactionException("errors");
  		}
  	}
  	
  	private void checkState(State shouldBe) throws TransactionException {
  		if (!stateRef.get().equals(shouldBe)) {
  			throwTransactionException("state is: "+stateRef.get()+" should be: "+shouldBe);
  		}
  	}
  	
  	private void check(State shouldBe, P participant) throws TransactionException {
  		checkParticipant(participant);
  		checkErrors();
  		checkState(shouldBe);
  	}
  	
  	private void setState(State state, State shouldBe) throws TransactionException {
  		
  	}
  	
  	private void checkParticipant(P participant) throws TransactionException {
  		if (participant == null) {
  			stateRef.set(State.FAILED);
  			throwTransactionException("participant null");
  		}
  		if (!participants.contains(participant)) {
  			stateRef.set(State.FAILED);
  		  throwTransactionException("unknown participant: "+participant);
  		}
  	}
  	
  	public void waitForPrepared(P participant, Timeout timeout) throws TransactionException, InterruptedException {
  		check(State.SUBMITTED, participant);
  		latch.countDown();
  		boolean worked = latch.await(timeout.getTimeLeft(), TimeUnit.MILLISECONDS);
  		if (!worked) {
  			stateRef.set(State.FAILED);
  			throwTransactionException("timed out");
  		}
  	}
  	
  	public void waitForCommit(P participant, Timeout timeout) throws TransactionException, InterruptedException {
  		check(State.PREPARED, participant);
  		boolean locked = lock.tryLock(timeout.getTimeLeft(), TimeUnit.MILLISECONDS);
  		if (!locked) {
  			throwTransactionException("timed out");
  		}
  		try {
  			commitCondition.await(timeout.getTimeLeft(), TimeUnit.MILLISECONDS);
  		} finally {
  			lock.unlock();
  		}
  	}
  	
  	public void doCommit(Timeout timeout) throws TransactionException, InterruptedException {
  		checkErrors();
  		checkState(State.SUBMITTED);
  		boolean awaited = latch.await(timeout.getTimeLeft(), TimeUnit.MILLISECONDS);
  		if (!awaited) {
  			throwTransactionException("timed out");
  		}
  		checkErrors();
  		checkState(State.SUBMITTED);
  		stateRef.set(State.PREPARED);
  		boolean locked = lock.tryLock(timeout.getTimeLeft(), TimeUnit.MILLISECONDS);
  		if (!locked) {
  			throwTransactionException("timed out");
  		}
  		try {
  			checkState(State.PREPARED);
  			checkErrors();
  			commitCondition.signalAll();
  		} finally {
  			lock.unlock();
  		}
  		if (!stateRef.compareAndSet(State.PREPARED, State.COMMITTED)) {
  			stateRef.set(State.FAILED);
  			throwTransactionException("error on commit");
  		}
  	}
  	
  	public State getState() {
  		return stateRef.get();
  	}
  }
}
